package com.fengye.middleware.rabbitmq.boot.config;

import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.AcknowledgeMode;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.config.SimpleRabbitListenerContainerFactory;
import org.springframework.amqp.rabbit.connection.CachingConnectionFactory;
import org.springframework.amqp.rabbit.connection.CorrelationData;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.amqp.support.converter.Jackson2JsonMessageConverter;
import org.springframework.amqp.support.converter.SimpleMessageConverter;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.autoconfigure.amqp.SimpleRabbitListenerContainerFactoryConfigurer;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

/**
 * RabbitMQ自定义注入相关组件
 * @author sky
 * @date 2021年06月09日 10:31 上午
 */
@Configuration
@Slf4j
public class RabbitMQConfig {

    @Autowired
    private CachingConnectionFactory cachingConnectionFactory;

    @Autowired
    private SimpleRabbitListenerContainerFactoryConfigurer factoryConfigurer;

    @Bean(name = "singleListenerContainer")
    public SimpleRabbitListenerContainerFactory listenerContainerFactory() {
        SimpleRabbitListenerContainerFactory factory = new SimpleRabbitListenerContainerFactory();
        factory.setConnectionFactory(cachingConnectionFactory);
        // 设置消息在传输中的格式，在这里采用 JSON 格式进行传输
        factory.setMessageConverter(new Jackson2JsonMessageConverter());

        // 设置并发消费者示例的初始数量
        factory.setConcurrentConsumers(1);

        // 设置并发消费者示例的最大数量
        factory.setMaxConcurrentConsumers(1);

        // 设置并发消费者示例中每个示例拉取的数量
        factory.setPrefetchCount(1);
        return factory;
    }

    @Bean(name = "stringListenerContainerFactory")
    public SimpleRabbitListenerContainerFactory stringListenerContainerFactory() {
        SimpleRabbitListenerContainerFactory factory = new SimpleRabbitListenerContainerFactory();
        factory.setConnectionFactory(cachingConnectionFactory);
        // 设置消息在传输中的格式，在这里采用 JSON 格式进行传输
        factory.setMessageConverter(new SimpleMessageConverter());

        // 设置并发消费者示例的初始数量
        factory.setConcurrentConsumers(1);

        // 设置并发消费者示例的最大数量
        factory.setMaxConcurrentConsumers(1);

        // 设置并发消费者示例中每个示例拉取的数量
        factory.setPrefetchCount(1);
        return factory;
    }

    @Bean(name = "multiListenerContainer")
    public SimpleRabbitListenerContainerFactory multiListenerContainerFactory() {
        SimpleRabbitListenerContainerFactory factory = new SimpleRabbitListenerContainerFactory();
        // factory.setConnectionFactory(cachingConnectionFactory);
        factoryConfigurer.configure(factory, cachingConnectionFactory);
        // 设置消息在传输中的格式，在这里采用 JSON 格式进行传输
        factory.setMessageConverter(new Jackson2JsonMessageConverter());

        // 设置消息的确认消费模式：AcknowledgeMode.NONE表示不需确认消费
        factory.setAcknowledgeMode(AcknowledgeMode.NONE);

        // 设置并发消费者示例的初始数量
        factory.setConcurrentConsumers(10);

        // 设置并发消费者示例的最大数量
        factory.setMaxConcurrentConsumers(15);

        // 设置并发消费者示例中每个示例拉取的数量
        factory.setPrefetchCount(10);
        return factory;
    }


    /**
     * 自定义配置 RabbitTemplate
     * @return org.springframework.amqp.rabbit.core.RabbitTemplate
     */
    @Bean
    public RabbitTemplate rabbitTemplate() {

        // 设置发送后进行确认
        // cachingConnectionFactory.setPublisherConfirms(true);
        cachingConnectionFactory.setPublisherConfirmType(CachingConnectionFactory.ConfirmType.CORRELATED);

        // 设置发送消息后返回确认消息
        cachingConnectionFactory.setPublisherReturns(true);

        // 构造消息组件实例
        RabbitTemplate rabbitTemplate = new RabbitTemplate(cachingConnectionFactory);

        // 发送消息时，发送成功的反馈信息
        rabbitTemplate.setConfirmCallback(new RabbitTemplate.ConfirmCallback() {
            @Override
            public void confirm(CorrelationData correlationData, boolean ack, String cause) {
                log.info("发送消息成功：correlationData({}),ack({}),cause({})", correlationData, ack, cause);
            }
        });

        // 发送消息时，发送失败的反馈信息
        rabbitTemplate.setReturnCallback(new RabbitTemplate.ReturnCallback() {
            @Override
            public void returnedMessage(Message message, int replyCode, String replyText, String exchange, String routingKey) {
                log.info("消息丢失：exchange({}),routingKey({}),replyCode({}),replyText({}),message({})",exchange,routingKey,replyCode,replyText,message);
            }
        });

        return rabbitTemplate;
    }
}
